#ifndef __MY_MYSQL_H__
#define __MY_MYSQL_H__


#include <mysql/mysql.h>
#include <mysql/mysqld_error.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>
#include <pthread.h>
#include <unistd.h>
#include "log.h"
#include "stdint.h"
#include "ringbuf_mpc.h"

struct easy_mysql_handle {
    char dbhost[64];
    char dbname[64];
    char dbpass[64];
    char dbuser[64];
    char last_error[512];
    char msg[512];
    int debug;
    int show_error;
    int reconnect_when_error;
    int commit_interval;
    int connected;
    uint64_t ignore_task_count;
    ringbuf_mpc_t *ringbuf;
    uint16_t dbport;
};

void reload_db_conf(int v);

#define easy_mysql_get_handle(dbhost, dbuser, dbpass, dbname, dbport, msg, must) ({ \
    MYSQL *mysql; int i; char *_buf = msg; \
    for (;;) { \
        for (i = 0; i < 3; i++) { mysql = mysql_init(NULL); if (mysql) break; sleep(1); } \
        if (mysql == NULL) { \
            LOGERR("mysql_init... fail\n"); \
        } else { \
            if (mysql_real_connect(mysql, dbhost, dbuser, dbpass, dbname, dbport, NULL, 0) == NULL) { \
               if (_buf) sprintf(_buf, "mysql connect(%s %u %s %s %s)... fail[%s]", dbhost, dbport, dbuser, dbpass, dbname, mysql_error(mysql)); \
               else           LOGERR("mysql connect(%s %u %s %s %s)... fail[%s]\n", dbhost, dbport, dbuser, dbpass, dbname, mysql_error(mysql)); \
               mysql_close(mysql); mysql = NULL; \
            } else { \
               if (_buf) sprintf(_buf, "mysql connect(%s %u %s %s %s)... ok", dbhost, dbport, dbuser, dbpass, dbname); \
           } \
        } \
        if (must == 0 || mysql) break; \
        sleep(5); \
        reload_db_conf(0); \
    } \
    mysql; })

static inline int easy_mysql_sql2int(const char *dbhost, const char *dbuser, const char *dbpass, const char *dbname, uint16_t dbport, const char *sql)
{
    int ret = -1;
    MYSQL *mysql = easy_mysql_get_handle(dbhost, dbuser, dbpass, dbname, dbport, NULL, 0);
    if (mysql == NULL) return -1;
    if (mysql_real_query(mysql, sql, strlen(sql))) { mysql_close(mysql); return -2; }
    MYSQL_RES *result = mysql_store_result(mysql);
    if (result) {
        MYSQL_ROW row = mysql_fetch_row(result);
        if (row) ret = atoi(row[0]);
        mysql_free_result(result);
    }
    mysql_close(mysql);
    return ret;
}

#define easy_mysql_sql2str(dbhost, dbuser, dbpass, dbname, dbport, mysql_handle, sql, buf) ({ \
    int _ret = 0; \
    buf[0] = 0; \
    MYSQL *_mysql_handle = mysql_handle; \
    if (_mysql_handle == NULL) _mysql_handle = easy_mysql_get_handle(dbhost, dbuser, dbpass, dbname, dbport, NULL, 0); \
    if (_mysql_handle == NULL) _ret = -1; \
    if (_ret == 0 && mysql_real_query(_mysql_handle, sql, strlen(sql))) { _ret = -2; } \
    if (_ret == 0) { \
        MYSQL_RES *result = mysql_store_result(_mysql_handle); \
        if (result) { \
            MYSQL_ROW row = mysql_fetch_row(result); \
            if (row == NULL) { \
                LOGFEW("mysql query error[%s]\n", sql); \
            } else { \
                if (row[0]) strcpy(buf, row[0]); \
            } \
            mysql_free_result(result); \
        } \
    } \
    if (mysql_handle == NULL && _mysql_handle) mysql_close(_mysql_handle); \
    _ret; })

static inline void *easy_mysql_work_thread(void *arg)
{
    MYSQL *mysql = NULL;
    struct easy_mysql_handle *handle = arg;
    time_t last_commit_ts = time(NULL);
    time_t last_query_ts  = time(NULL);
    uint32_t last_ping_query_count = 0;
    uint32_t query_count = 0;
    uint32_t diff;
    char value = 1;
    char *sql;

    pthread_detach(pthread_self());
    LOGINF("%s up\n", __func__);

reconnect:
    handle->connected = 0;
    if (mysql) { mysql_close(mysql); mysql = NULL; }
    mysql = easy_mysql_get_handle(handle->dbhost, handle->dbuser, handle->dbpass, handle->dbname, handle->dbport, handle->msg, 1); if (mysql == NULL) { sleep(3); goto reconnect; }
    handle->connected = 1;
    if (handle->commit_interval) mysql_autocommit(mysql, 0);
    mysql_options(mysql, MYSQL_OPT_RECONNECT, (char*)&value);
    for (;;) {
        time_t ts = time(NULL);
        if (handle->commit_interval) {
            diff = ts - last_commit_ts;
            if (diff >= handle->commit_interval) {
                mysql_commit(mysql); last_commit_ts = ts;
            }
        }
        diff = ts - last_query_ts;
        if (diff >= 20) {
            if (last_ping_query_count == query_count) {
                mysql_ping(mysql); last_ping_query_count = query_count; last_query_ts = ts;
            }
        }

        if (ringbuf_mpc_get_total_data_len(handle->ringbuf) < sizeof(void *)) {
            struct timespec ts; clock_gettime(CLOCK_REALTIME, &ts); ts.tv_sec += 1;
            pthread_mutex_lock(&handle->ringbuf->c_lock);
            pthread_cond_timedwait(&handle->ringbuf->c_cond, &handle->ringbuf->c_lock, &ts);
            pthread_mutex_unlock(&handle->ringbuf->c_lock);
            continue;
        }
        ringbuf_mpc_get_data(handle->ringbuf, &sql, sizeof(void *), 1);
        if (handle->debug > 0) LLOG("exe sql task %s\n", sql);
        int ret = mysql_real_query(mysql, sql, strlen(sql)); query_count++;
        if (ret) {
            sprintf(handle->last_error, "fail to query[%s][ret %d][%s]", sql, ret, mysql_error(mysql));
            if      (handle->debug > 0      ) LLOG("%s\n", handle->last_error);
            else if (handle->show_error >= 2) LLOG("%s\n", handle->last_error);
            else if (handle->show_error >= 1) LOGFEW("%s\n", handle->last_error);
            if (handle->reconnect_when_error > 0) { free(sql); goto reconnect; }
        } else {
            if (handle->debug > 0) LLOG("mysql query[%s] done\n", sql);
        }
        free(sql);
    }
    goto reconnect;
    return NULL;
}

static inline struct easy_mysql_handle *easy_mysql_init(const char *dbhost, uint16_t dbport, const char *dbuser, const char *dbpass, const char *dbname, int debug, int reconnect_when_error, int show_error, int commit_interval)
{
    struct easy_mysql_handle *handle = calloc(1, sizeof(*handle)); if (handle == NULL) { LOGERR("fail to alloc\n"); return NULL; }

    handle->ringbuf = create_ringbuf_mpc(4096 * sizeof(void *));
    strcpy(handle->dbhost, dbhost);
    strcpy(handle->dbname, dbname);
    strcpy(handle->dbuser, dbuser);
    strcpy(handle->dbpass, dbpass);
    handle->dbport = dbport;
    handle->debug = debug;
    handle->reconnect_when_error = reconnect_when_error;
    handle->show_error = show_error;
    handle->commit_interval = commit_interval;
    pthread_t pid;
    pthread_create(&pid, NULL, easy_mysql_work_thread, handle);
    return handle;
}

#define easy_mysql_add_task(handle, sql, force) ({ int ret = 0; \
        if (handle->debug > 0) LLOG("add sql task %s\n", *sql); \
        if (ringbuf_mpc_add_data(handle->ringbuf, sql, sizeof(void *), force)) { free(*(sql)); handle->ignore_task_count++; ret = -1; } \
        pthread_cond_signal(&handle->ringbuf->p_cond); \
        ret; })
#endif
